package com.gzyj.flink.gps.map;

import com.gzyj.flink.dto.alarm.Kafka_VehicleAreaAlarm;
import com.gzyj.flink.gps.GPSPosition;
import com.gzyj.flink.gps.map.area.AreaAlarmDataProcess;
import com.gzyj.flink.gps.map.area.AreaAlarmResult;
import com.gzyj.flink.gps.sink.gp.BaobiaoAreaalarmSink;
import com.gzyj.flink.pojo.gps.BaobiaoAreaalarm;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import java.util.Objects;

public class AreaAlarmFlow {

    private SingleOutputStreamOperator<GPSPosition> gpsSingleStream;

    public AreaAlarmFlow(SingleOutputStreamOperator<GPSPosition> gpsSingleStream) {
        this.gpsSingleStream = gpsSingleStream;
    }

    public void process() {

        SingleOutputStreamOperator<AreaAlarmResult> areaAlarmResultSingleOutputStream = this.gpsSingleStream.flatMap(new AreaAlarmDataProcess()).filter(Objects::nonNull);


        areaAlarmResultSingleOutputStream.filter(e -> e.getKafkaVehicleAreaAlarm() != null)
                .map(e -> e.getKafkaVehicleAreaAlarm())
                .map(new MapFunction<Kafka_VehicleAreaAlarm, BaobiaoAreaalarm>() {
                    @Override
                    public BaobiaoAreaalarm map(Kafka_VehicleAreaAlarm VehicleAreaAlarm) throws Exception {
                        BaobiaoAreaalarm baobiaoAreaalarm=new BaobiaoAreaalarm();
                        baobiaoAreaalarm.setVehicleid(VehicleAreaAlarm.getVehicleId());
                        baobiaoAreaalarm.setIsacross(VehicleAreaAlarm.isIsAcross()?1:0);
                        baobiaoAreaalarm.setNote(VehicleAreaAlarm.getNote());
                        baobiaoAreaalarm.setTime(VehicleAreaAlarm.getTime());
                        baobiaoAreaalarm.setVelocity(VehicleAreaAlarm.getVelocity());
                        baobiaoAreaalarm.setAreaid(VehicleAreaAlarm.getAreaId());
                        baobiaoAreaalarm.setTimes(VehicleAreaAlarm.getTimes());
                        baobiaoAreaalarm.setGuid(VehicleAreaAlarm.getAlarmGuid());
                        baobiaoAreaalarm.setCreatetime(VehicleAreaAlarm.getCreateTime());
                        baobiaoAreaalarm.setLongitude(VehicleAreaAlarm.getLongitude());
                        baobiaoAreaalarm.setLatitude(VehicleAreaAlarm.getLatitude());
                        baobiaoAreaalarm.setRoadname(VehicleAreaAlarm.getRoadName());

                        return baobiaoAreaalarm;
                    }
                }).addSink(new BaobiaoAreaalarmSink());


        areaAlarmResultSingleOutputStream
                .filter(e -> e.getKafkaVehicleRemindContent() != null)
                .map(e -> e.getKafkaVehicleRemindContent())
                .print();

    }
}
